package com.xiaoyu.im.client;

import in.srain.cube.concurrent.AppThreads;

import com.xiaoyu.im.IMCallback;
import com.xiaoyu.im.concurrent.MessageQueueProcessor;
import com.xiaoyu.im.datamodel.vendors.VendorConversationUnique;
import com.xiaoyu.im.provider.AbstractMessageDataProvider;
import com.xiaoyu.base.data.UserData;
import com.xiaoyu.im.event.IMOnlineStatusUpdateEvent;
import com.xiaoyu.im.event.OnlineUserAccountEvent;
import com.xiaoyu.im.datamodel.common.IMConstant;
import com.xiaoyu.im.datamodel.messagesync.MessageFetchResult;
import com.xiaoyu.im.client.proxy.IMProxy;
import com.xiaoyu.im.client.proxy.nim.AntiSpamResult;
import com.xiaoyu.im.client.proxy.nim.NimProxy;
import com.xiaoyu.im.datamodel.AppMessage;
import com.xiaoyu.im.provider.MessageStatusProcessor;
import com.xiaoyu.im.request.MessageRequest;
import com.xiaoyu.im.request.flow.RequestFlow;
import com.xiaoyu.base.utils.ListUtil;

import java.util.Collections;
import java.util.List;

import in.srain.cube.util.CLog;
import in.srain.cube.util.internal.AppConsumer2;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;

import static com.xiaoyu.im.datamodel.messagesync.MessageFetchResult.COMPLETE;
import static com.xiaoyu.im.datamodel.messagesync.MessageFetchResult.HAS_MORE;
import static com.xiaoyu.im.client.proxy.AbstractProxy.QueryDirection.NEW;

public class IMClient {

    private static final IMClient INSTANCE = new IMClient();

    private final IMProxy mIMProxy;
    private final MessageQueueProcessor<MessageRequest> mMessageRequestQueue;
    private final MessageQueueProcessor<AppMessage> mMessageReceiveQueue;
    private AbstractMessageDataProvider mMessageDataProvider;

    public static IMClient getInstance() {
        return INSTANCE;
    }

    private IMClient() {
        mIMProxy = NimProxy.getInstance();
        mMessageRequestQueue = new MessageRequestQueue();
        mMessageReceiveQueue = new MessageReceiveQueue();
    }

    public void init(AbstractMessageDataProvider messageDataProvider, AppConsumer2<String, Long> receiptFunction) {
        mMessageDataProvider = messageDataProvider;

        mIMProxy.setUp();
        mIMProxy.observeReceiveMessage(this::receiveMessages);  //消息接收
        mIMProxy.observeCustomNotification(this::receiveCustomNotification);  //自定义通知接收
        mIMProxy.observeRecallMessage(mMessageDataProvider::onMessageRecallAsync);  //消息撤回
        mIMProxy.observeOnlineUserAccounts(accountSet -> new OnlineUserAccountEvent(accountSet).post()); //监听用户订阅的事件
        mIMProxy.observeBroadcastMessage(mMessageDataProvider::onBroadcastMessage); //IM系统消息广播
        mIMProxy.observeMessageReceipt(appMessageReceipts -> mMessageDataProvider.onMessageReceipt(appMessageReceipts, receiptFunction)); //消息已读
        mIMProxy.observeOnlineStatus(status -> {
            //监听用户在线状态
            new IMOnlineStatusUpdateEvent(status).postSticky();
        });
    }

    public void login() {
        if (UserData.getInstance().hasLogin()) {
            mIMProxy.login();
        }
    }

    public void logout() {
        mIMProxy.logout();
    }

    public String getVendorType() {
        return mIMProxy.getVendorType();
    }

    public void sendMessage(AppMessage message, IMCallback<AppMessage> callback) {
        mIMProxy.sendMessage(message, callback);
    }

    public void sendMessageReceipt(AppMessage message) {
        mIMProxy.sendMessageReceipt(message);
    }

    public void receiveMessages(List<AppMessage> messages) {
        ListUtil.forEach(messages, mMessageReceiveQueue::enqueueMessage);
    }

    public void receiveCustomNotification(List<AppMessage> messages) {
        ListUtil.forEach(messages, mMessageReceiveQueue::enqueueMessage);
    }

    public void recallMessage(AppMessage message, IMCallback<Void> callback) {
        mIMProxy.recallMessage(message, callback);
    }

    public int getUnreadMessageCount(String chatId) {
        return mIMProxy.getUnreadMessageCount(chatId);
    }

    public void clearUnreadCount(String chatId) {
        mIMProxy.clearUnreadCount(chatId);
    }

    public boolean isOnline() {
        return mIMProxy.isOnline();
    }

    public String getOnlineStatus() {
        return mIMProxy.getOnlineStatus();
    }

    public AntiSpamResult checkLocalAntiSpam(String content, String replacement) {
        return mIMProxy.checkLocalAntiSpam(content, replacement);
    }

    public void enqueueMessageRequest(MessageRequest request) {
        CLog.d(IMConstant.LOG_TAG, "enqueueMessageRequest: %s", request);
        MessageStatusProcessor.getInstance().addSending(request.getLocalId());
        AppThreads.runOnDBThread(() -> mMessageDataProvider.preparePendingMessage(request));
        mMessageRequestQueue.enqueueMessage(request);
    }

    public Observable<MessageFetchResult> fetchRoamMessage(String chatId, long fromTime, long toTime) {
        return Observable.create(emitter -> {
            try {
                MessageFetchResult fetchResult = mIMProxy.onFetch(chatId, fromTime, toTime, NEW);
                fetchFinish(emitter, chatId, fetchResult);
            } catch (Throwable throwable) {
                emitter.onError(throwable);
            }
        });
    }

    public void fetchFinish(ObservableEmitter<MessageFetchResult> emitter, String chatId, MessageFetchResult fetchResult) {
        switch (fetchResult.status) {
            case HAS_MORE:
                mIMProxy.onFetchFinish(chatId, fetchResult.messages);
                emitter.onNext(fetchResult);
                break;
            case COMPLETE:
                emitter.onComplete();
                break;
        }
    }

    public VendorConversationUnique getConversationUnique(String chatId) {
        return mIMProxy.toConversationUnique(chatId);
    }

    private static class MessageRequestQueue extends MessageQueueProcessor<MessageRequest> {

        private static final String NAME = IMConstant.LOG_TAG + "-out";

        private MessageRequestQueue() {
            super(NAME);
        }

        @Override
        protected void processMessage(MessageRequest request) {
            RequestFlow requestFlow = new RequestFlow();
            requestFlow.start(request);
        }
    }

    private class MessageReceiveQueue extends MessageQueueProcessor<AppMessage> {

        private static final String NAME = IMConstant.LOG_TAG + "-in";

        private MessageReceiveQueue() {
            super(NAME);
        }

        @Override
        protected void processMessage(AppMessage message) {
            mMessageDataProvider.onMessageReceived(message);
        }
    }
}
